package com.dianshang.utils;

import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedLongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * ESClient
 * 操作ES的客户端对象
 */
public class ESClient {

    private String url; // IP地址
    private Integer port; // 端口号
    private RestHighLevelClient restHighLevelClient; // 连接ES的客户端对象

    public ESClient(){}

    public ESClient(String url,Integer port) {
        this.url = url;
        this.port = port;
        this.restHighLevelClient =  restHighLevelClient();
    }

    /**
     * 获取一个操作ES的客户端对象
     */
    private RestHighLevelClient restHighLevelClient(){
        HttpHost httpHost = new HttpHost(url,port);
        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(restClientBuilder);
        return restHighLevelClient;
    }

    /**
     * 获取一个查询的请求对象
     */
    public SearchRequest searchRequest(String indexName,String typeName){
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.types(typeName);
        return searchRequest;
    }

    /**
     * 获取返回的结果
     */
    public List<Map> getResultMap(SearchResponse searchResponse){
        SearchHit[] hits = searchResponse.getHits().getHits();
        List<Map> result = new ArrayList();
        for (SearchHit searchHit : hits){
            Map<String, Object> sourceAsMap = searchHit.getSourceAsMap(); // Map结果
            // 获取name 高亮的值
            Map<String, HighlightField> highlightFields = searchHit.getHighlightFields();
            HighlightField highLightName = highlightFields.get("name");
            sourceAsMap.put("name",highLightName.getFragments()[0].toString());
            result.add(sourceAsMap);
        }
        return result;
    }

    /**
     * 查询符合条件的总的条数
     */
    public long getTotal(SearchResponse searchResponse){
       return (long) searchResponse.getHits().getTotalHits();
    }

    /**
     * 获取返回的结果
     */
    public List<String> getResultJson(SearchResponse searchResponse){
        SearchHit[] hits = searchResponse.getHits().getHits();
        List<String> result = new ArrayList();
        for (SearchHit searchHit : hits){
            String sourceAsString = searchHit.getSourceAsString(); // Map结果
            result.add(sourceAsString);
        }
        return result;
    }

    /**
     * 获取分组的结果
     */
    public List<Map> getAgg(SearchResponse searchResponse,String aggName){
        Aggregations aggregations = searchResponse.getAggregations(); // 获取分组的结果
        Map<String, Aggregation> stringAggregationMap = aggregations.asMap(); // 分组结果 使用Map形式来表示
        ParsedLongTerms ageresult = (ParsedLongTerms) stringAggregationMap.get(aggName); // 根据起的别名,获取分组的最终的结果
        List<? extends Terms.Bucket> buckets = ageresult.getBuckets(); // 分组结果  Bucket：有具体分组名称,分组总数
        List<Map> aggMap = new ArrayList<Map>();
        for (Terms.Bucket bucket : buckets){
            Object key = bucket.getKey(); // 获取分组名称
            long docCount = bucket.getDocCount(); // 获取当前分组的总数
            HashMap map = new HashMap();
            map.put("name",key);
            map.put("count",docCount);
            aggMap.add(map);
        }
        return aggMap;
    }

    /**
     * 获取所有分组名称集合
     */
    public List getAggKeys(SearchResponse searchResponse,String aggName){
        Aggregations aggregations = searchResponse.getAggregations(); // 获取分组的结果
        Map<String, Aggregation> stringAggregationMap = aggregations.asMap(); // 分组结果 使用Map形式来表示
        ParsedTerms ageresult = (ParsedTerms) stringAggregationMap.get(aggName); // 根据起的别名，获取分组的最终的结果
        List<? extends Terms.Bucket> buckets = ageresult.getBuckets(); // 分组结果 Bucket：有具体分组名称，分组总数
        List aggList = new ArrayList();
        for (Terms.Bucket bucket : buckets){
            Object key = bucket.getKey(); // 获取分组名称
            aggList.add(key);
        }
        return aggList;
    }

    /**
     *  批量新增文档
     */
    public void addBlukDoc(String indexName,String typeName,List<String> docList){
        System.out.println(Thread.currentThread().getName()+"--docList:"+docList.size());
        BulkRequest bulkRequest=new BulkRequest();
        for (String doc : docList){
            IndexRequest indexRequest = new IndexRequest(indexName,typeName);
            indexRequest.source(doc, XContentType.JSON);
            bulkRequest.add(indexRequest);
        }
        // 触发访问ES
        try {
            restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public RestHighLevelClient getRestHighLevelClient() {
        return restHighLevelClient;
    }

    public void setRestHighLevelClient(RestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
    }

}